package logreport

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"go.uber.org/zap"
	"gobase/logger"
	"gobase/rocketmq"
)

var defaultReport *Report

var Topics = []string{
	"sys", //系统环境
	"log",
	"broker-a",
	"service",
}
var Groups = []string{
	"test_service",
}

//var lags = []string{
//	"测试",
//	"log",
//}

type Report struct {
	prods       map[string]*rocketmq.MqProducer
	tag         string
	group       string
	callback    func(ctx context.Context, result *primitive.SendResult, err error)
	NameSrvAddr []string
	topic       map[string]string
}

//func init() {
//
//	defaultReport = NewReport()
//
//	defaultHook()
//}

func GetGroup(topic string) string {
	switch topic {
	case "sys":
		return "test_service"
	default:
		return "test_service"
	}
}

func NewReport(NameSrvAddr []string) *Report {
	prods := make(map[string]*rocketmq.MqProducer, 6)

	for _, t := range Topics {
		prods[GetGroup(t)] = rocketmq.NewMQProducer(t, GetGroup(t), NameSrvAddr)
	}

	return &Report{
		prods: prods,
		topic: make(map[string]string),
	}
}

func (r *Report) Tag() string {
	return r.tag
}

func (r *Report) Group() string {
	return r.group
}

func (r *Report) SetTag(tag string) {
	r.tag = tag
}

func (r *Report) SetGroup(group string) {
	r.group = group
}

func (r *Report) SetHook(hook func(ctx context.Context, result *primitive.SendResult, err error)) {
	r.callback = hook
}

func (r *Report) SendSyncMessage(msg *rocketmq.MQMessage) error {
	prod, ok := r.prods[GetGroup(msg.TopicName)]
	if !ok {
		err := fmt.Errorf("topic %s not exist", msg.TopicName)
		logger.Error("Topic Not Found", zap.String("topic", msg.TopicName))
		return err
	}
	if _, err := prod.SendSyncMessage(msg); err != nil {
		return err
	}
	return nil
}

func (r *Report) SendAsyncMessage(msg *rocketmq.MQMessage) error {
	prod, ok := r.prods[GetGroup(msg.TopicName)]
	if !ok {
		err := fmt.Errorf("topic %s not exist", msg.TopicName)
		logger.Error("Topic Not Found", zap.String("topic", msg.TopicName))
		return err
	}
	if _, err := prod.SendAsyncMessage(msg, r.callback); err != nil {
		return err
	}
	return nil
}

func defaultHook() {
	defaultReport.SetHook(func(ctx context.Context, result *primitive.SendResult, err error) {
		if err != nil {
			logger.Error("SendAsyncMessage Error", zap.Error(err))
			return
		}
		select {
		case <-ctx.Done():
			logger.Warn("stop with context done")
			return
		default:
			logger.Info("SendAsyncMessage Success", zap.Any("result", result))
		}
	})
}

func DefaultGroup(group string) {
	defaultReport.SetGroup(group)
}

func DefaultHook(hook func(ctx context.Context, result *primitive.SendResult, err error)) {
	defaultReport.SetHook(hook)
}

// 发送异步消息后回调
func SendAsyncMessage(msg *rocketmq.MQMessage) error {
	return defaultReport.SendAsyncMessage(msg)
}

// 发送同步消息
// TODO 有高并发bug
func SendSyncMessage(msg *rocketmq.MQMessage) error {
	return defaultReport.SendSyncMessage(msg)
}
